package io.xxx.sync.core

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.JSONObject
import com.alibaba.fastjson.JSONPath
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper
import io.xxx.sync.dao.*
import org.quartz.DisallowConcurrentExecution
import org.quartz.Job
import org.quartz.JobExecutionContext
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.web.client.RestTemplateBuilder
import org.springframework.core.io.Resource
import org.springframework.http.HttpEntity
import org.springframework.http.HttpHeaders
import org.springframework.http.HttpMethod
import org.springframework.http.MediaType
import org.springframework.http.client.OkHttp3ClientHttpRequestFactory
import org.springframework.web.client.RestTemplate
import org.springframework.web.util.UriComponentsBuilder
import java.io.BufferedReader
import java.io.InputStreamReader
import java.time.Duration
import java.time.LocalDateTime
import java.util.concurrent.LinkedBlockingQueue
import java.util.concurrent.TimeUnit
import kotlin.concurrent.thread


/**
 * 根据[SyncSchedule]拉取数据保存到[SyncDocument]中并更新[SyncSchedule]。
 * @see ScheduleCalendar
 */
@Suppress("SpringJavaAutowiredMembersInspection")
@DisallowConcurrentExecution
open class Synchronizer(protected var property: SyncProperty) : Job {

    @Autowired
    private lateinit var authInfoMapper: AuthInfoMapper

    @Autowired
    private lateinit var propertyMapper: SyncPropertyMapper

    @Autowired
    private lateinit var scheduleMapper: SyncScheduleMapper

    @Autowired
    protected lateinit var documentMapper: SyncDocumentMapper

    @Autowired
    private lateinit var restTemplateBuilder: RestTemplateBuilder

    protected val restTemplate: RestTemplate by lazy {
        restTemplateBuilder
            .requestFactory(OkHttp3ClientHttpRequestFactory::class.java)
            .build()
    }

    override fun execute(context: JobExecutionContext) {
        acquire()

        val property = propertyMapper.selectById(property.id) // refresh property
        if (property == null) {
            if (log.isInfoEnabled) {
                context.apply {
                    val jobKey = jobDetail.key
                    scheduler.deleteJob(jobKey)
                    log.info("同步器${this@Synchronizer.property.id}配置不存在${if (scheduler.checkExists(jobKey)) "，任务已删除" else ""}")
                }
            }
            return
        } else {
            this.property = property
        }

        fun pullAndSave(parameter: Any? = null) {
            fun execute(schedules: List<SyncSchedule>, parameter: Any?) {
                if (schedules.isEmpty()) return

                val st = System.currentTimeMillis()
                val schedule = if (schedules.size > 1) {
                    val sortedSchedules = schedules.sortedBy { it.startTime }
                    sortedSchedules.last().copy(startTime = schedules.first().startTime)
                } else {
                    schedules[0]
                }

                try {
                    val wrapper = QueryWrapper<AuthInfo>()
                        .eq("channel", property.channel)
                    val authInfos = authInfoMapper.selectList(wrapper)
                    if (authInfos.isEmpty()) {
                        pullAndSave(null, schedule, parameter)
                    } else {
                        var stream = authInfos.stream()
                        if (parallel) {
                            stream = stream.parallel()
                        }
                        stream.forEach { pullAndSave(it, schedule, parameter) }
                    }
                } catch (e: Exception) {
                    with(property) {
                        log.error("[$id](${schedule.id})数据同步发生异常", e)
                    }
                }

                if (finished()) {
                    scheduleMapper.updateAll(schedules.onEach {
                        it.completed = true
                    })
                }
                if (log.isTraceEnabled) {
                    log.trace("${property.id}数据同步完成，耗时: ${System.currentTimeMillis() - st}ms")
                }
            }

            val wrapper = QueryWrapper<SyncSchedule>()
                .eq("property_id", property.id)
                .eq("completed", false)
                .orderByDesc("priority")
                .orderByAsc("start_time")
                .last("limit ${duration.seconds / 60}")
            val uncompletedSchedules = scheduleMapper.selectList(wrapper)
            if (uncompletedSchedules.isNotEmpty()) {
                uncompletedSchedules.groupBy { it.priority }.forEach {
                    execute(it.value, parameter)
                }
            }
        }

        if (parameters.isEmpty()) pullAndSave()
        else parameters.forEach { pullAndSave(it) }
    }

    /**
     * 门店是否可以独立的运行在不同的线程中，默认为true。
     */
    protected open var parallel = true

    /**
     * 合并schedule的时间跨度，默认为1天，防止数据量过大出现OOM。
     * 根据平台的开放文档中的开始时间和结束时间最大范围进行配置。
     */
    protected open var duration = Duration.ofDays(1)!!

    /**
     * 扩展参数由子类提供并共享给父类方法。
     */
    protected open val parameters: Array<Any> = emptyArray()

    /**
     * 阻塞程序继续运行知道满足条件，如限流。
     */
    protected open fun acquire() {}

    /**
     * 根据条件来终止Schedule，true表示Schedule可以标记为完成，false表示Schedule还不可以标记为已完成。
     */
    protected open fun finished(): Boolean {
        return true
    }

    /**
     * 拉取并保存数据，Synchronizer提供了基于response-stream的实现，PageSynchronizer提供了基于分页的实现，
     * 数据来源分为内部渠道和外部渠道，对于内部渠道的数据同步，直接配置SyncProperty就可以使用了；
     * 但是外部渠道不支持response-stream模式，基本上需要通过覆盖此方法完成基于分页模式的同步，
     * 继承[PageSynchronizer]，重新实现[PageSynchronizer.getCount]和[PageSynchronizer.getData]方法。
     * @param authInfo  授权信息
     * @param schedule  时间片
     * @param parameter 扩展参数，一般用于外部渠道的一些特殊参数
     * @return 数据总数
     */
    protected open fun pullAndSave(authInfo: AuthInfo?, schedule: SyncSchedule, parameter: Any?): Long {
        val url = getQueryUrl(property.urlPath!!, schedule)
        val response = getResponse<Resource>(url) ?: return 0
        val thread = thread(name = "SAVE_DOCS_THREAD") {
            while (!done) {
                queue.poll(1, TimeUnit.SECONDS)?.apply {
                    documentMapper.insertAll(this)
                }
            }
        }

        var count = 0L
        try {
            val br = BufferedReader(InputStreamReader(response.inputStream), 8192 * 100)
            val buffer = mutableListOf<SyncDocument>()
            br.forEachLine {
                val document = buildDocument(JSON.parseObject(it))
                buffer += document
                if (buffer.size == property.pageSize) {
                    queue.put(buffer.toList())
                    buffer.clear()
                }
                count++
            }
            if (buffer.isNotEmpty()) queue.put(buffer)
        } finally {
            done = true
        }
        thread.join()
        return count
    }

    protected inline fun <reified T> getResponse(
        url: String,
        method: HttpMethod = HttpMethod.GET,
        requestEntity: HttpEntity<*>? = null
    ): T? {
        try {
            return restTemplate.exchange(url, method, requestEntity, T::class.java).body
        } catch (e: Exception) {
            log.error(
                "接口调用失败，URL: ${method}:${url}"
            )
            throw e
        }
    }

    fun headers(): HttpHeaders {
        val headers = HttpHeaders()
        headers.contentType = MediaType.APPLICATION_JSON
        if (!property.tokenName.isNullOrBlank()) {
            headers[property.tokenName!!] = property.tokenValue
        }
        return headers
    }

    protected fun getQueryUrl(
        path: String,
        schedule: SyncSchedule,
        params: Map<String, Any> = emptyMap()
    ): String {
        val builder = with(property) {
            val builder = UriComponentsBuilder.fromHttpUrl(httpUrl!!)
                .path(path)
            urlQueryParams?.split(",")?.forEach {
                val kv = it.split("=")
                builder.queryParam(kv[0], kv[1])
            }
            builder
        }

        return with(schedule) {
            params.forEach { (name, value) ->
                builder.replaceQueryParam(name, value)
            }
            builder.replaceQueryParam("shopCode", shopCode)
                .replaceQueryParam("propertyId", propertyId)
                .replaceQueryParam("startTime", startTimeValue)
                .replaceQueryParam("endTime", endTimeValue)
                .toUriString()
        }
    }

    protected fun buildDocument(jsonObject: JSONObject): SyncDocument =
        with(property) {
            val sn = getValue<String>(jsonObject, snJsonPath)
            val rsn = getValue<String>(jsonObject, rsnJsonPath)
            val status = getValue<String>(jsonObject, statusJsonPath)
            val receiverName = getValue<String>(jsonObject, receiverNameJsonPath)
            val receiverMobile = getValue<String>(jsonObject, receiverMobileJsonPath)
            val created = getValue<LocalDateTime>(jsonObject, createdJsonPath)
            val modified = getValue<LocalDateTime>(jsonObject, modifiedJsonPath)
            SyncDocument(
                sn!!,
                rsn,
                status,
                receiverName,
                receiverMobile,
                created,
                modified,
                JSON.toJSONString(jsonObject)
            )
        }

    companion object {
        val log: Logger = LoggerFactory.getLogger(Synchronizer::class.java)

        @Volatile
        private var done = false
        private val queue = LinkedBlockingQueue<List<SyncDocument>>(500)

        private inline fun <reified T> getValue(jsonObject: JSONObject, path: String?): T? {
            if (path.isNullOrBlank()) {
                return null
            }
            val value = JSONPath.eval(jsonObject, path) ?: return null
            return when (T::class.java) {
                LocalDateTime::class.java -> LocalDateTime.parse(value.toString(), formatter) as T
                else -> value.toString() as T
            }
        }

        fun throwException(api: String?, code: Any?, msg: String?) {
            throw RuntimeException("${api}: ${code}[${msg}]")
        }
    }
}

/**
 * 通过分页的方式同步数据
 */
@Suppress("SpringJavaAutowiredMembersInspection")
open class PageSynchronizer(property: SyncProperty) : Synchronizer(property) {

    /**
     * 默认数据起始页码为1
     */
    protected open val startPage by lazy {
        property.startPage
    }

    /**
     * 默认每页100条数据
     */
    protected open val pageSize = 100

    /**
     * 返回数据总数
     * @param authInfo  授权信息
     * @param schedule  时间片
     * @param parameter 扩展参数，一般用于外部渠道的一些特殊参数
     */
    protected open fun getCount(authInfo: AuthInfo?, schedule: SyncSchedule, parameter: Any?): Long? {
        val url = getQueryUrl(property.countUrlPath, schedule)
        return if (property.dataJsonPath.isNullOrBlank()) {
            getResponse<Long>(url) ?: throw RuntimeException("[${schedule.id}]获取数据发生异常")
        } else {    // ex: {"code": 0, message: "success", data: 0}
            val response = getResponse<JSONObject>(url)
                ?: throw RuntimeException("[${schedule.id}]获取数据发生异常")
            try {
                JSONPath.eval(response, property.dataJsonPath).toString().toLong()
            } catch (e: Exception) {
                log.error("解析数据发生异常", e)
                throw e
            }
        }
    }

    /**
     * 返回数据对象，需要将原始数据包装成[SyncDocument]，根据实际情况设置辅助字段。
     * 如果modified未被设置，则数据不会进行更新。
     * @param authInfo  授权信息
     * @param schedule  时间片
     * @param parameter 扩展参数，一般用于外部渠道的一些特殊参数
     */
    protected open fun getData(
        authInfo: AuthInfo?,
        schedule: SyncSchedule,
        parameter: Any?,
        pageNo: Long
    ): List<SyncDocument> {
        val url = getQueryUrl(
            property.dataUrlPath, schedule, mapOf("pageSize" to pageSize, "pageNumber" to pageNo)
        )
        return if (property.dataJsonPath.isNullOrBlank()) {
            val response = getResponse<List<JSONObject>>(url)
                ?: throw RuntimeException("[${schedule.id}]获取数据发生异常")
            response.map { buildDocument(it) }
        } else {    // ex: {"code": 0, message: "success", data: []}
            val response = getResponse<JSONObject>(url)
                ?: throw RuntimeException("[${schedule.id}]获取数据发生异常")
            try {
                @Suppress("UNCHECKED_CAST")
                val jsonArray = JSONPath.eval(response, property.dataJsonPath) as List<Map<String, Any>>
                jsonArray.map { buildDocument(JSONObject(it)) }
            } catch (e: Exception) {
                log.error("解析数据发生异常", e)
                throw e
            }
        }
    }

    protected open fun saveData(document: SyncDocument) {
        val wrapper = QueryWrapper<SyncDocument>()
            .select("id", "modified")
            .eq("property_id", document.propertyId)
            .eq("sn", document.sn)
        val oldDoc = documentMapper.selectOne(wrapper)
        document.apply {
            if (oldDoc == null) {
                documentMapper.insert(this)
            } else if ((modified != null && oldDoc.modified == null)
                || (modified != null && oldDoc.modified != null && modified!! > oldDoc.modified)
            ) {
                id = oldDoc.id
                documentMapper.updateById(this)
            }
        }
    }

    protected open fun saveData(documents: List<SyncDocument>) {
        if (documents.isEmpty()) return

        val wrapper = QueryWrapper<SyncDocument>()
            .select("id", "sn", "modified")
            .eq("property_id", property.id)
            .`in`("sn", documents.map { it.sn })
        val oldDocMap = documentMapper.selectList(wrapper).associateBy({ it.sn }, { it })
        val (newDocs, oldDocs) = documents.partition { !oldDocMap.containsKey(it.sn) }
        val modifiedDocs = oldDocs.filter {
            with(it) {
                val oldDoc = oldDocMap.getValue(it.sn)
                id = oldDoc.id
                (modified != null && oldDoc.modified == null) || (modified != null && oldDoc.modified != null && modified!! > oldDoc.modified)
            }
        }

        if (newDocs.isNotEmpty()) documentMapper.insertAll(newDocs)
        if (modifiedDocs.isNotEmpty()) documentMapper.updateAll(modifiedDocs)
    }

    // 分页模式的标准实现
    final override fun pullAndSave(authInfo: AuthInfo?, schedule: SyncSchedule, parameter: Any?): Long {
        val count = getCount(authInfo, schedule, parameter)
        var pages = if (count == null) 0 else (count / pageSize + if (count % pageSize == 0L) 0 else 1)
        while (pages-- > 0) {  // 页码从0开始
            val data = getData(authInfo, schedule, parameter, pages + startPage).onEach {
                it.propertyId = property.id
                it.shopCode = authInfo?.targetShopCode
            }

            if (property.batched) {
                saveData(data)
            } else {
                data.parallelStream().forEach { saveData(it) }
            }
        }
        return count ?: 0
    }
}